package org.chen;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.Set;
import java.util.UUID;

public class DelayMsgQueue {
  private Jedis jedis;
  private String queue;

  public DelayMsgQueue(Jedis jedis, String queue) {
    this.jedis = jedis;
    this.queue = queue;
  }

  /**
   * 消息入队
   * @param data 要发送的消息
   */
  public void queue(Object data){
    //将要发送的消息内容封装到QueueMessage实体类对象中
    QueueMessage queueMessage = new QueueMessage();
    queueMessage.setId(UUID.randomUUID().toString());
    queueMessage.setData(data);

    //序列化消息对象
    try {
      String value = new ObjectMapper().writeValueAsString(queueMessage);
      System.out.println("消息发送:" + new Date());

      //消息发送，score=现在时间+延迟5秒
      jedis.zadd(queue,System.currentTimeMillis()+5000,value);

    } catch (JsonProcessingException e) {
      e.printStackTrace();
    }
  }

  /**
   * 消费消息
   */
  public void loop(){
    //轮询，如果线程没被中断，则读取消息
    while (!Thread.interrupted()){
      //获取score范围为0～当前时间的消息的第一条
      Set<String> zrange = jedis.zrangeByScore(queue, 0, System.currentTimeMillis(), 0, 1);

      if (zrange.isEmpty()){
        //如果消息队列为空
        //等待7秒后继续读取消息队列
        try {
          Thread.sleep(7000);
        } catch (InterruptedException e) {
          break;
        }
        continue;
      }

      //如果读取到了queue中的消息，则直接读取消息出来
      String next = zrange.iterator().next();
      if (jedis.zrem(queue,next)>0){
        //读取到消息了，开始业务操作

          //解析收到的消息
        try {
          QueueMessage message = new ObjectMapper().readValue(next, QueueMessage.class);
          System.out.println(new Date()+" 收到消息:" + message.toString());
        } catch (JsonProcessingException e) {
          e.printStackTrace();
        }
      }

    }
  }
}
